package com.mq.mqserver;

import com.mq.common.*;
import com.mq.common.Arguments.*;
import com.mq.common.exception.MqException;
import com.mq.mqserver.core.BasicProperties;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/*
这个BrokerServer就是 消息队列 的主题
本质上是一个TCP 服务器*/

public class BrokerServer {
    // 当前程序只考虑⼀个虚拟主机的情况.
    private VirtualHost virtualHost = new VirtualHost("default");
    // 存储当前会话，有哪些客户端和服务器在通信
    // key 为 channelId, value 为 channel 对应的 socket 对象.
    private ConcurrentHashMap<String, Socket> sessions = new ConcurrentHashMap<>();
    //serverSocket 是服务器⾃⾝的 socket
    private ServerSocket serverSocket = null;
    //executorService 这个线程池⽤来处理响应，处理多客户端请求
    private ExecutorService executorService;
    //runnable 这个标志位⽤来控制服务器的运⾏停⽌，volatile：变量的更新操作通知到其他线程
    private volatile boolean runnable = true;

    //TCP服务器，启动/停⽌服务器
    public BrokerServer(int port) throws IOException {
        serverSocket = new ServerSocket(port);//绑定端口
    }
    public void start() throws IOException {
        System.out.println("[BrokerServer] 启动完成!");
        executorService = Executors.newCachedThreadPool();
        try {
            //volatile修饰runnable，就是为了，runnable状态改变，这里能立马检查到
            while (runnable) {
                Socket clientSocket = serverSocket.accept();
                //把处理连接的逻辑丢给线程池
                executorService.submit(() -> processConnection(clientSocket));
            }
        } catch (SocketException e) {
            System.out.println("[BrokerServer] 服务器关闭完成!");
        }
    }
    //一般停止服务器就直接 kill 掉对应的进程
    //这里做一个测试方法，主要用于后续的单元测试
    public void stop() throws IOException {
        runnable = false;
        // ⽴即结束所有的线程池的任务
        executorService.shutdownNow();
        serverSocket.close();
    }
    //通过这个方法处理一个客户端的连接
    //在一个连接中可能会涉及到多个请求与响应
    private void processConnection(Socket clientSocket) {
        try (InputStream inputStream = clientSocket.getInputStream();
             OutputStream outputStream = clientSocket.getOutputStream()) {
            //这里需要按照特定格式读取并解析，此时需要用到 DataInputSream 和 DataOutputSream
            try(DataInputStream dataInputStream = new DataInputStream(inputStream);
            DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                while (true) {
                    //1.读取请求并解析
                    Request request = readRequest(dataInputStream);
                    //2.根据请求计算响应
                    Response response = process(request, clientSocket);
                    //3.把响应写回客户端
                    writeResponse(dataOutputStream, response);
                }
            }
        } catch (EOFException | SocketException e) {
            //DataInputStream 如果读到 EOF，就会抛出EOFException 异常
            //需要借助这个关闭异常
            System.out.println("[BrokerServer] connection 关闭! serverIP=" +
                    clientSocket.getInetAddress().toString() + ", port=" + clientSocket.getPort());
        } catch (IOException | ClassNotFoundException | MqException e) {
            System.out.println("[BrokerServer] connection 出现异常!");
            e.printStackTrace();
        } finally {
            try {
                //连接处理完就需要关闭 socket
                clientSocket.close();
                // 一个TCP连接中，包含多个channel,需要把当前这个 socket 对应的所有 channel 清理掉
                // 对 sessions 进⾏清理
                clearClosedSession(clientSocket);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    //遍历上述 sessions hash表，把被关闭的 socket 对应的键值对都删掉
    //可能出现channel还没有来得及关闭，tcp连接结束了
    private void clearClosedSession(Socket clientSocket) {
        // 这⾥不要在同⼀个循环中, 同时进⾏遍历 + 删除操作. 否则可能有迭代器失效问题.
        // 拆成两个循环来处理是更合适的.
        List<String> toDeleteChannelId = new ArrayList<>();
        for (Map.Entry<String, Socket> entry : sessions.entrySet()) {
            if (entry.getValue() == clientSocket) {
                //集合类不能一边遍历一边删除，会导致迭代器失效
                //sessions.remove(entry.getKey());
                //先记录要删除的ChannelId
                toDeleteChannelId.add(entry.getKey());
            }
        }
        for (String channelId : toDeleteChannelId) {
            //遍历List集合类，删除Map的元素
            sessions.remove(channelId);
        }
        System.out.println("[BrokerServer] 清理 session 完成! 被清理的 channelId=" + toDeleteChannelId);

    }


    private void writeResponse(DataOutputStream dataOutputStream, Response response) throws IOException {
        dataOutputStream.writeInt(response.getType());//写4个字节是Type
        dataOutputStream.writeInt(response.getLength());//再写四个字节是Length
        dataOutputStream.write(response.getPayload());//再写 Payload
        //刷新缓冲区，保证数据快速进入流中
        dataOutputStream.flush();

    }
    private Request readRequest(DataInputStream dataInputStream) throws IOException {
        Request request = new Request();
        request.setType(dataInputStream.readInt());//读4个字节是Type
        request.setLength(dataInputStream.readInt());//再读四个字节是Length
        byte[] payload = new byte[request.getLength()];//再读多少字节取决于Length
        int n = dataInputStream.read(payload);
        //长度和预期不符合，抛异常
        if (n != request.getLength()) {
            throw new IOException("读取请求数据出错!");
        }
        request.setPayload(payload);
        return request;
    }
    private Response process(Request request, Socket clientSocket) throws IOException, ClassNotFoundException, MqException {
        // 1. 从 request 中解析出业务请求
        //BasicArguments是所有type方法的父类，因此先把类型转化为BasicArguments
        BasicArguments baseArguments = (BasicArguments) BinaryTool.fromBytes(request.getPayload());
        System.out.println("[Request] rid=" + baseArguments.getRid() + ", channelId=" + baseArguments.getChannelId()
                + ", type=" + request.getType() + ", length=" + request.getLength());
        // 2. 根据 type 来区分业务分支
        boolean ok = true;
        if (request.getType() == 0x1) {
            // 创建 channel
            sessions.put(baseArguments.getChannelId(), clientSocket);
            System.out.println("[BrokerServer] 创建 channel 完成! channelId=" +
                    baseArguments.getChannelId());
        } else if (request.getType() == 0x2) {
            // 销毁 channel
            sessions.remove(baseArguments.getChannelId());
            System.out.println("[BrokerServer] 销毁 channel 完成! channelId=" +
                    baseArguments.getChannelId());
        } else if (request.getType() == 0x3) {
            // 创建交换机 ，payload就是 ExchangeDeclareArguments 的属性
            ExchangeDeclareArguments exchangeDeclareArguments =
                    (ExchangeDeclareArguments) baseArguments;
            ok = virtualHost.exchangeDeclare(exchangeDeclareArguments.getExchangeName(),
                    exchangeDeclareArguments.getExchangeType(),
                    exchangeDeclareArguments.isDurable(),
                    exchangeDeclareArguments.isAutoDelete(),
                    exchangeDeclareArguments.getArguments());
        } else if (request.getType() == 0x4) {
            // 删除交换机
            ExchangeDeleteArguments exchangeDeleteArguments =
                    (ExchangeDeleteArguments) baseArguments;
            ok = virtualHost.exchangeDelete(exchangeDeleteArguments.getExchangeName());
        } else if (request.getType() == 0x5) {
            // 创建队列
            QueueDeclareArguments queueDeclareArguments = (QueueDeclareArguments)
                    baseArguments;
            ok = virtualHost.queueDeclare(queueDeclareArguments.getQueueName(),
                    queueDeclareArguments.isDurable(), queueDeclareArguments.isExclusive(), queueDeclareArguments.isAutoDelete(),
                    queueDeclareArguments.getArguments());
        } else if (request.getType() == 0x6) {
            // 删除队列
            QueueDeleteArguments queueDeleteArguments = (QueueDeleteArguments)
                    baseArguments;
            ok = virtualHost.queueDelete(queueDeleteArguments.getQueueName());
        } else if (request.getType() == 0x7) {
            // 创建绑定
            QueueBindArguments queueBindArguments = (QueueBindArguments)
                    baseArguments;
            ok = virtualHost.queueBind(queueBindArguments.getQueueName(),
                    queueBindArguments.getExchangeName(), queueBindArguments.getBindingKey());
        } else if (request.getType() == 0x8) {
            // 解除绑定
            QueueUnbindArguments queueUnbindArguments = (QueueUnbindArguments)
                    baseArguments;
            ok = virtualHost.queueUnbind(queueUnbindArguments.getQueueName(),
                    queueUnbindArguments.getExchangeName());
        } else if (request.getType() == 0x9) {
            // 发送消息
            BasicPublishArguments basicPublishArguments = (BasicPublishArguments)
                    baseArguments;
            ok = virtualHost.basicPublish(basicPublishArguments.getExchangeName(),
                    basicPublishArguments.getRoutingKey(),
                    basicPublishArguments.getBasicProperties(),
                    basicPublishArguments.getBody());
        } else if (request.getType() == 0xa) {
            // 订阅消息
            BasicConsumeArguments basicConsumeArguments = (BasicConsumeArguments) baseArguments;
            // 创建个回调, ⽤来把消费的数据转发回客⼾端.
            ok = virtualHost.basicConsumer(basicConsumeArguments.getConsumeTag(), basicConsumeArguments.getQueueName(),
                    basicConsumeArguments.isAutoAck(), new Consumer() {
                        @Override
                        public void handleDelivery(String consumerTag,
                             BasicProperties properties, byte[] body) throws MqException, IOException {
                            // 1. 根据 channelId 找到对应的 socket
                            //此处 consumerTag 是 channelId ，根据 channelId 去 sessions 中查询，
                            // 就可以得到对应的socket对象，从而可以发数据
                            Socket clientSocket =
                                    sessions.get(basicConsumeArguments.getChannelId());
                            if (clientSocket == null || clientSocket.isClosed()) {
                                throw new MqException("[BrokerServer] 订阅消息的客户端已经关闭!");
                            }
                            // 2. 构造响应数据
                            //不仅自身，继承的属性也要设定
                            SubScribeReturns subScribeReturns = new SubScribeReturns();
                            subScribeReturns.setChannelId(basicConsumeArguments.getChannelId());
                            subScribeReturns.setRid(""); //这里响应没有请求，不需要去对应，rid暂时不需要
                            subScribeReturns.setOk(true);
                            subScribeReturns.setConsumerTag(consumerTag);
                            subScribeReturns.setBasicProperties(properties);
                            subScribeReturns.setBody(body);
                            //序列化成字节数组
                            byte[] payload = BinaryTool.toBytes(subScribeReturns);
                            // 3. 写⼊到对应 socket 中
                            Response response = new Response();
                            //0xc 表示服务器给消费者客户端推送的消息数据
                            response.setType(0xc);
                            response.setLength(payload.length);
                            response.setPayload(payload);
                            // 此处不应该关闭 DataOutputStream,
                            // 关闭这个会导致内部持有的 clientSocket.getOutputStream 被关闭.
                            //此时无法继续往 socket中写入后续数据
                            DataOutputStream dataOutputStream = new
                                    DataOutputStream(clientSocket.getOutputStream());
                            writeResponse(dataOutputStream, response);
                        }
                    });
        } else if (request.getType() == 0xb) {
            // 调用BasicAck，确认消息
            BasicAckArguments basicAckArguments = (BasicAckArguments) baseArguments;
            ok = virtualHost.basicAck(basicAckArguments.getQueueName(),
                    basicAckArguments.getMessageId());
        } else {
            throw new MqException("[BrokerServer] 未知的请求 type ! type=" + request.getType());
        }
        // 3. 构造响应.
        BasicReturns basicReturns = new BasicReturns();
        //和请求一致
        basicReturns.setRid(baseArguments.getRid());
        basicReturns.setChannelId(baseArguments.getChannelId());
        basicReturns.setOk(ok);
        byte[] payload = BinaryTool.toBytes(basicReturns);

        Response response = new Response();
        response.setType(request.getType());
        response.setLength(payload.length);
        response.setPayload(payload);
        System.out.println("[Response] rid=" + basicReturns.getRid() + ", channelId=" + basicReturns.getChannelId()
                        + ", type=" + response.getType() + ", length=" + response.getLength());
        return response;
    }

}

